All committers have signed the CLA.

⚠️ No Changeset found

Latest commit: d01e684

Merging this PR will not cause a version bump for any packages. If these changes should not result in a new version, you're good to go. If these changes should result in a version bump, you need to add a changeset.

Click here to learn what changesets are, and how to add one.

Click here if you're a maintainer who wants to add a changeset to this PR

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

Hey @mshivam019, the DeferredReadableStream is intentionally designed to prevent calling setSource after detachSource. Here's why:

Consider an async task reading from this stream:

const deferred = new DeferredReadableStream<string>();
const reader = deferred.stream.getReader();

while (true) {
  const { value, done } = await reader.read();
  if (done) break;
  // process value...
}

Once detachSource is called, done becomes true and the read loop exits. This follows the standard ReadableStream contract—when done is true, the stream is fully closed and consumers should expect no further values. Re-attaching a source would violate this expectation.

Regarding your scenarios:

• Browser reload: The previous room connection should be disconnected with no reconnection expected.
• Network interruption: This triggers reconnecting events, but the stream itself shouldn't need re-attachment.

Could you clarify: have you observed this deferred readable stream blocking the agent's progress during reconnection? If so, could you share more details about the specific scenario? That would help us better understand if there's an edge case we need to address.

@toubatbrian Thanks for the detailed explanation of the design rationale!

Let me clarify my specific use case:

Scenario: Interview Session with Reconnection Grace Period

I'm building a voice AI interview application where users may accidentally:

• Reload the page or navigate away briefly and return
• Experience browser/device crashes

For this, I want a 3-minute grace window where the user can rejoin the same room and continue the conversation with full context preserved.

How I've configured this:

Room creation (backend):

await roomService.createRoom({
  name: roomName,
  metadata: roomMetadata,
  emptyTimeout: 300,      // 5 minutes before any participant joins
  departureTimeout: 180,  // 3 minutes grace period after last participant leaves
  maxParticipants: 10,
});

Agent session (keeps session alive when user disconnects):

await session.start({
  agent: assistant,
  room: ctx.room,
  inputOptions: {
    noiseCancellation: BackgroundVoiceCancellation(),
    closeOnDisconnect: false,  // Keep agent session alive for reconnection
  },
});

The Problem

When the user reconnects (same room, same identity):

1. LiveKit fires TrackSubscribed for their new microphone track
2. ParticipantAudioInputStream.onTrackSubscribed() calls closeStream() → detachSource()
3. Then immediately calls setSource(newAudioStream)
4. Without this patch: Throws "Stream source already set"
5. Result: Agent cannot hear user's audio (one-way audio - user hears agent, agent doesn't hear user)

Question

If making DeferredReadableStream reusable isn't the preferred approach, what would you recommend for this "reconnection with context preservation" scenario?

Should ParticipantAudioInputStream perhaps create a new DeferredReadableStream instance on reconnection rather than reusing the existing one? Or is there another pattern you'd suggest?

Happy to adjust the implementation based on your guidance!

Hey @mshivam019, I made a PR to support a new stream primitive, which could hopefully support the usecase you want: #1036.

Feel free to test on that branch and let me know if you encounter any issues!

Hey @toubatbrian, thanks for the pointer! In the meantime, I migrated my architecture to create a fresh room on reconnection instead — the backend preloads the conversation context into the new agent session, so the user picks up right where they left off without needing stream reuse. That eliminated the need for this PR on my end.

That said, keeping the original room open would still be the better approach — with the new room architecture, there's a window where recent conversation data that's still in the agent's context but hasn't been persisted by the backend yet can get lost during the handoff. The stream reuse approach avoids that entirely since the session stays alive.

I checked out #1036 and the deferred stream approach looks solid. I'll give it a try soon and report back if I run into anything.

I'm closing this PR since we've merged #1036